package com.demo;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;

import java.net.InetSocketAddress;
import java.util.List;

/**
 * @Description:
 * @Author: dengping.xu
 * @Data: 2021-11-03
 */
public class CanalPartsDemo {

    private static final String HOST_IP = "10.118.72.6";
    private static final int PORT = 63306;
    private static final int BATCH_SIZE = 120;

    public static void main(String[] args) {
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(HOST_IP, PORT), "parts", "", "");
        connector.connect();
        connector.subscribe("");
        connector.rollback();
        while(true){
            Message message = connector.getWithoutAck(BATCH_SIZE);
            long id = message.getId();
            int size = message.getEntries().size();
            if(id == -1 || size ==0){
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }else {
                HandleMessages(message.getEntries());
            }
            connector.ack(id);
        }
    }

    private static void HandleMessages(List<CanalEntry.Entry> entries) {
        for (CanalEntry.Entry entry : entries) {
            if(entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
                    || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND){
                continue;
            }
            CanalEntry.RowChange rowChange;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());

                System.out.println(String.format("================》; binlog[%s:%s] , name[%s,%s] , eventType : %s",
                        entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                        entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                        rowChange.getEventType()));
                if(rowChange.getIsDdl()){
                    System.out.println("================》;isDdl: true,sql:" + rowChange.getSql());
                }

                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                    if(CanalEntry.EventType.DELETE == rowChange.getEventType()){
//                        partsService.deleteById(Integer.valueOf(rowData.getProps(0).getValue()));
                        /*for (CanalEntry.Column column : rowData.getBeforeColumnsList()) {
                            if(column.getName().equals("id")){
                                partsService.deleteById(Integer.valueOf(column.getValue()));
                            }
                        }*/

                    }else if(CanalEntry.EventType.INSERT == rowChange.getEventType()){

                    }else {

                    }
                }
            } catch (InvalidProtocolBufferException e) {
                e.printStackTrace();
            }
        }
    }
}
